package com.fanxuankai.boot.canal.listener.autoconfigure;

import com.fanxuankai.boot.canal.listener.MessageListener;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.consumer.EventListener;
import com.fanxuankai.boot.mqbroker.consumer.EventListenerBean;
import com.fanxuankai.boot.mqbroker.consumer.EventListenerContainer;
import com.fanxuankai.boot.mqbroker.producer.EventPublisher;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;

import java.util.ArrayList;
import java.util.List;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({EventPublisher.class})
@ConditionalOnProperty(name = CanalListenerProperties.ENABLED, matchIfMissing = true)
public class CanalListenerToEventListenerAutoConfiguration {
    @Bean
    public EventListenerContainer eventListenerContainer(MessageListener messageListener) {
        EventListenerContainer container = new EventListenerContainer();
        List<EventListenerBean> listeners = new ArrayList<>();
        for (String topic : messageListener.getTopics()) {
            // 不能使用 lambda, 否则无法获取泛型
            EventListener<String> eventListener = new EventListener<String>() {
                @Override
                public void onEvent(Event<String> event) {
                    messageListener.onMessage(event.getName(), event.getData());
                }
            };
            listeners.add(new EventListenerBean(topic, eventListener));
        }
        container.setListeners(listeners);
        return container;
    }
}
